// @Author EthanScriptOn
// @Desc
package kafka_operator

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator/op_example"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

type KafkaConsumeOperator struct {
	*op_example.LoopDetectionMultiChannelDataOperator
	*identity.IdentityInformationCarrier
}

func (k *KafkaConsumeOperator) ExecuteWithLoop(ctx *actuator.SrfSwitchBridgeActuatorContext, operatorConfigIdentifier interface{}, structureParam interface{}) (err error) {
	operatorIdentifier, err := k.IdentityInformationCarrier.GenerateUniqueIdentifier()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error generating unique identifier: %v", err))
		return err
	}
	pipeline := k.GetDataPipeline(operatorIdentifier)
	if pipeline == nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting data pipeline: %v", err))
		return err
	}
	consumer, err := connect.GetKafkaConsume()
	if err != nil {
		return err
	}
	config := operator.GetOperatorConfig(operatorConfigIdentifier)
	if config == nil {
		return fmt.Errorf("config is nil")
	}
	operatorConfig, ok := config.(*extend_config.KafkaConsumeOperatorConfig)
	if !ok {
		return fmt.Errorf("config is not *KafkaConsumeOperatorConfig")
	}
	err = consumer.SubscribeTopics(operatorConfig.Topics, nil)
	if err != nil {
		return err
	}
	for {
		ev := consumer.Poll(1000)
		if ev == nil {
			continue
		}
		switch e := ev.(type) {
		case *kafka.Message:
			_, err = consumer.CommitMessage(e)
			if err != nil {
				_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error committing message: %v", err))
				pipeline <- e
				continue
			}
			pipeline <- e
		case kafka.Error:
			_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Received message with exception: %v", e))
			pipeline <- e
		default:
			_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Unknown message type: %v", e))
			pipeline <- e
		}
	}
}
